// Copyright (C) 2012 Yury Bychenkov (http://bychenkov.com).
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//  - Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//  - Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//  - Neither the name of Yury Bychenkov nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

#include "nunchunk.h"

Logger NunChunk::logger("NunChunk");

NunChunk::NunChunk(MappedFile* mappedFile_):
	mappedFile(mappedFile_),
	header((const Header*) mappedFile->data()),
	data(((const uint8*) header) + header->headerSize)
{
	dictionaries = (Dictionary*) new uint8[header->columnCount * sizeof(Dictionary)];
	indecies = (Index*) new uint8[header->columnCount * sizeof(Index)];

	for (uint columnIndex = 0; columnIndex < header->columnCount; columnIndex++) {
		const ColumnEntry& entry = header->columns[columnIndex];

		const uint8* columnData = data + entry.dataOffset;
		                                             
		new (dictionaries + columnIndex) Dictionary(columnData + entry.dictionaryOffset, entry.dictionarySize, entry.dictionaryCount, (const uint32*) (columnData + entry.dictionarySkipTableOffset), entry.dictionarySkipStep);
		new (indecies + columnIndex) Index(columnData + entry.indexOffset, entry.indexSize, entry.indexChunkSize);
	}                                                
}

NunChunk::~NunChunk() {
	close();
}


bool NunChunk::isOpened() const {
	return mappedFile != 0;
}

void NunChunk::close() {
	if (mappedFile != 0) {
		for (uint columnIndex = 0; columnIndex < header->columnCount; columnIndex++) {
			dictionaries[columnIndex].~Dictionary();
			indecies[columnIndex].~Index();
		}
		
		delete [] (uint8*) dictionaries;
		dictionaries = 0;
		
		delete [] (uint8*) indecies;
		indecies = 0;

		delete mappedFile;
		mappedFile = 0;
		data = 0;
		header = 0;
	}
}

NunChunk* NunChunk::open(const char* file) {
	Header header;

	ifstream is(file, ios::binary | ios::in);
	if (!is.is_open()) {
		logger.error("File ", file, " doesn't exist");
		return 0;
	}
	if (sizeof(Header) != is.read((char*) &header, sizeof(Header)).gcount()) {
		logger.error("Can't load the header: ", file);
		return 0;
	}
	is.close();

	if (header.magic != MAGIC) {
		logger.error("MAGIC mismatch: ", file);
		return 0;
	}

	MappedFile* mappedFile = MappedFile::open(file, 0, true);
	if (mappedFile == 0) {
		logger.error("Can't map file ", file , "to memory");
		return 0;
	}

	return new NunChunk(mappedFile);
}


bool NunChunk::build(const Schema& schema, const char* file, const LogChunk& chunk) {
	MappedFile* mf = MappedFile::open(file, 0x40000000ULL, false);
	if (mf == 0) {
		return 0;
	}
	uint8* data = mf->data();

	uint columnCount = schema.getColumnCount();

	uint headerSize = (sizeof(Header) + sizeof(ColumnEntry) * (columnCount - 1) + 4095) & ~4095;
	memset(data, 0, headerSize);

	Header* header = (Header*) data;
	header->magic = MAGIC;
	header->headerSize = headerSize;
	header->columnCount = columnCount;
	header->rowCount = 0;
	header->rowSkipStep = 16;

	uint8* columnData = data + headerSize;

	for (uint columnIndex = 0; columnIndex < columnCount; columnIndex++) {
		header->columns[columnIndex].dataOffset = (uint64)(columnData - data - headerSize);
		uint type = schema.getColumn(columnIndex).type;
		ColumnEntry& entry = header->columns[columnIndex];

		entry.type = (uint32) type;

		uint size;
		if (type == Schema::STRING) {
			size = addColumn<StringTraits>(*header, entry, columnData, chunk, columnIndex);
		} else if (type == Schema::INTEGER) {
			size = addColumn<IntegerTraits>(*header, entry, columnData, chunk, columnIndex);
		} else if (type == Schema::DECIMAL) {
			size = addColumn<DecimalTraits>(*header, entry, columnData, chunk, columnIndex);
		} else {
			mf->close(0);
			return false;
		}

		if (size == 0) {
			mf->close(0);
			return false;
		}

		size = (size + 4095) & ~4095;
		columnData += size;
	}

	uint64 size = (columnData - data + sizeof(uint64) + 4095) & ~4095;

	mf->close(size);

	return true;
}


template <typename ElementTraits>
uint NunChunk::addColumn(Header& header, ColumnEntry& entry, uint8* data, const LogChunk& chunk, uint columnIndex) {

	// Raw data
	uint32 dataSize = 0;
	uint32 dataCount = 0;
	for (LogChunk::Iterator it = chunk.iterator(columnIndex); it.skip<ElementTraits>(); ) {
		uint elementSize = ElementTraits::sizeOf(it.getValue());
		memcpy(data + dataSize, it.getValue(), elementSize);
		dataSize += elementSize;
		dataCount++;
	}
	entry.dataSize = dataSize;
	if (header.rowCount == 0) {
		header.rowCount = dataCount;
	} else if (header.rowCount != dataCount) {
		stringstream ss;
		ss << "Row count doesn't match for column #" << columnIndex << ": " << dataCount << " (!= " << header.rowCount << " for previous columns)";
		logger.error(ss.str());
		return 0;
	}

	// Skip table for data
	entry.dataSkipTableOffset = (dataSize + 7) & ~7;
	uint skipStep = header.rowSkipStep;
	uint32* skipTable = (uint32*)(data + entry.dataSkipTableOffset);

	*skipTable++ = 0;
	const uint8* p = data;
	for (uint i = 0, k = 0; i < dataCount; i++, k++) {
		if (k == skipStep) {
			*skipTable++ = (uint32) (p - data);		
			k = 0;
		}

		p += ElementTraits::sizeOf(p);
	}


	// Dictionary
	entry.dictionaryOffset = (uint32) ((entry.dataSkipTableOffset + dataCount * sizeof(uint32) + 4095) & ~4095);

	uint memorySize = dataSize * 2 + 1024 * 1024;
	uint8* memory = new uint8 [memorySize];


	DictionaryBuilder<ElementTraits> dictionaryBuilder(memory, memorySize, 0, 0);

	for (LogChunk::Iterator it = chunk.iterator(columnIndex); it.skip<ElementTraits>(); ) {
		if (!dictionaryBuilder.add(it.getValue())) {
			delete [] memory;
			logger.error("Internal error: dictionary overflow");
			return 0;
		}
	}

	uint count = 0;
	uint64 size = 0;

	if (!dictionaryBuilder.build(data + entry.dictionaryOffset, memorySize, size, count)) {
	    delete [] memory;
		logger.error("Internal error: can't build dictionary");
		return 0;
	}

	delete [] memory;

	if (size <= DICTIONARY_SIZE_THRESHOLD) {
		memmove(data, data + entry.dictionaryOffset, size);
		entry.dictionaryOffset = 0;
		entry.dataSize = 0;
		entry.dataSkipTableOffset = 0;
	}

	entry.dictionarySize = (uint32) size;
	entry.dictionaryCount = (uint32) count;

	// Skip table for dictionary
	entry.dictionarySkipTableOffset = entry.dictionaryOffset + (uint32)((entry.dictionarySize + 7) & ~7);

	uint dictionarySkipStep = (uint32) (count / 65536);
	if (dictionarySkipStep < 1) {
		dictionarySkipStep = 1;
	} else if (dictionarySkipStep > 16) {
		dictionarySkipStep = 16;
	}
    uint dictionarySkipCount = (count - 1) / dictionarySkipStep + 1;
	entry.dictionarySkipStep = dictionarySkipStep;

	uint32* dictionarySkipTable = (uint32*)(data + entry.dictionarySkipTableOffset);

	*dictionarySkipTable++ = 0;

	uint8* dictionaryData = data + entry.dictionaryOffset;
	p = dictionaryData;
	for (uint i = 0, k = 0; i < count; i++, k++) {
		if (k == dictionarySkipStep) {
			*dictionarySkipTable++ = (uint32) (p - dictionaryData);
			k = 0;
		}

		p += ElementTraits::sizeOf(p);
	}



	// Index
   	entry.indexOffset = (uint32) ((entry.dictionarySkipTableOffset + dictionarySkipCount * sizeof(uint32) + 4095) & ~4095);
	entry.indexChunkSize = 256;

	Dictionary dictionary(data + entry.dictionaryOffset, entry.dictionarySize, entry.dictionaryCount, (uint32*)(data + entry.dictionarySkipTableOffset), dictionarySkipStep);

	IndexBuilder indexBuilder(data + entry.indexOffset, 8 * dataCount + 256, entry.indexChunkSize);

	for (LogChunk::Iterator it = chunk.iterator(columnIndex); it.skip<ElementTraits>(); ) {
		int k = dictionary.find<ElementTraits>(it.getValue());
		if (k < 0) {
			logger.error("Internal error: element not found ", dictionarySkipStep);
			return 0;
		}
		if (!indexBuilder.add(k)) {
			logger.error("Internal error: index overflow");
			return 0;
		}
	}

	entry.indexSize = (uint32) indexBuilder.build();

	return (uint)(entry.indexOffset + entry.indexSize);
}
